package org.example

import org.apache.spark.rdd.RDD
import org.apache.spark.sql.SparkSession
import org.apache.spark.sql.types.{IntegerType, StringType, StructField, StructType}

import java.util.Properties

object sparkYun_JDBC {
  def main(args: Array[String]): Unit = {
    val spark = SparkSession
      .builder()
      .master("local[*]")
      .appName("spark")
      .getOrCreate()
    val sc = spark.sparkContext
//    连接MySQL数据库
    val properties: Properties = new Properties();
    properties.setProperty("user","root")
    properties.setProperty("password","123456")
    properties.setProperty("driver","com.mysql.jdbc.Driver")//8.0版本要加.cj
    val mysqlScore = spark.read.jdbc("jdbc:mysql://localhost:3306/test?" +
      "verifyServerCertificate=false&useSSL=false","spark",properties)
    mysqlScore.createTempView("spark")
    //使用Navicat导入平时成绩，在spark中读取并计算平均分
//    val res1 = spark.sql(
//      """
//        |select
//        |name,
//        |avg(score) as avg_score
//        |from spark
//        |group by name
//        |""".stripMargin
//    )
//    res1.show()
//    //    往MySQL中写入数据  写入行  写入列
//    val data: RDD[String] = sc.makeRDD(Array("张三,1001,100","李四,1002,99"))
//    //    1.按MySQL列名切分数据
//    val dataRDD = data.map(_.split(","))
//    //    2.匹配样例类
//    val scoreRDD = dataRDD.map(x => Score(x(0),x(1),x(2)))
//    //    3.将RDD转换成DataFrame
//    import  spark.implicits._
//    val dataDF = scoreRDD.toDF()
//    dataDF.write.mode("append").jdbc("jdbc:mysql://localhost:3306/test?" +
//      "verifyServerCertificate=false&useSSL=false", "spark", properties)
    //    读取平时成绩csv文件并写入MySQL数据库的spark成绩中
    val schemaUser = StructType(Seq(
      StructField("name", IntegerType, nullable = true),
      StructField("number", IntegerType, nullable = false),
      StructField("score", IntegerType, nullable = false),
    ))
    val a = spark.read.option("sep", "::").schema(schemaUser)
      .csv("src/main/resources/yun2.csv")
    a.show()

    mysqlScore.show()
    sc.stop()
  }
  case class Score(name:String,number:String,score:String)
}
